package source;

import beans.SenSorReading;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Random;

public class SourceUDF01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<SenSorReading> dataStreamSource = env.addSource(new MySensorSource());

        dataStreamSource.print();
        env.execute();
    }
    //实现自定义的SourceFunction
    public static class MySensorSource implements SourceFunction<SenSorReading> {

        //第一一个标识位，用来控制数据的产生
        private boolean running = true;

        @Override
        public void run(SourceContext<SenSorReading> sourceContext) throws Exception {
            //第一一个随机数发生器
            Random random = new Random();

            //设置10个初始温度
            HashMap<String, Double> sensorTempMap = new HashMap<>();
            for (int i = 0; i < 10; i++) {
                sensorTempMap.put("sensor_" + (i + 1), 60 + random.nextGaussian() * 20);
            }

            while (running) {
                for (String sensorId : sensorTempMap.keySet()) {
                    //在前点温度基础上做一个随机波动
                    Double newtemp = sensorTempMap.get(sensorId) + random.nextGaussian();
                    sensorTempMap.put(sensorId, newtemp);
                    sourceContext.collect(new SenSorReading(sensorId, System.currentTimeMillis(), newtemp));
                }
            }
            //控制更新频率
            Thread.sleep(1000);
        }
        @Override
        public void cancel() {
            running = false;
        }
    }
}
